package com.czk.springboot.reactive.reactivestream;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo {
    public static void main(String[] args) {
        // 1 定义一个发布者
        // SubmissionPublisher发布的所有数据在buffer区
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        // 2 定义一个订阅者，订阅者·订阅发布者的数据
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {

            /**
             * d订阅时 在xxx事件发生时，执行回调
             *
             * @param subscription a new subscription
             */
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread().getName() + " 订阅开始了：" + subscription);
                // 从上游请求数据---可以指定请求数量  被背压模式，自己控制自己的索要的元素数量
                subscription.request(Long.MAX_VALUE);
            }

            /**
             * 下一个元素到达时，执行回调，接收新数据
             *
             * @param item the item
             */
            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread().getName() + " 接收到数据：" + item);
            }

            /**
             * 错误发生时执行回掉
             *
             * @param throwable the exception
             */
            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread().getName() + " 接收到错误信号：" + throwable);

            }

            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread().getName() + " 接收到完成信号：");
            }
        };

        // 绑定订阅者和发布者
        // jvm底层对于整个发布订阅关系做好了异步+缓冲区处理=响应式系统
        publisher.subscribe(subscriber);
        for (int i = 0; i < 10; i++) {
            publisher.submit(String.valueOf(i));
        }
        // 发布者关闭，订阅者回接收到完成信号
        publisher.close();
        while (true) {
        }
    }
}
